package com.atguigu.edu.app.dws;

import bean.TrafficHourPageViewBean;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import util.DateFormatUtil;
import util.MyClickhouseUtil;
import util.MyKafkaUtil;

import java.time.Duration;

public class DwsTrafficHourPageViewWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //TODO 2.检查点相关设置
        //TODO 3.从kafka的页面主题和独立访客主题读取数据
        String pageLogTopic = "dwd_traffic_page_log";
        String uvTopic = "dwd_traffic_unique_visitor_detail";
        String groupId = "dws_hour_page_view_group";
        DataStreamSource<String> pageLogStrDS = env.addSource(MyKafkaUtil.getKafkaConsumer(pageLogTopic, groupId));
        DataStreamSource<String> uvStrDS = env.addSource(MyKafkaUtil.getKafkaConsumer(uvTopic, groupId));
        //TODO 4.将两条流的数据类型转为 实体类对象
        //页面
        SingleOutputStreamOperator<TrafficHourPageViewBean> pageViewBeanDS = pageLogStrDS.process(new ProcessFunction<String, TrafficHourPageViewBean>() {

            @Override
            public void processElement(String jsonStr, ProcessFunction<String, TrafficHourPageViewBean>.Context ctx, Collector<TrafficHourPageViewBean> out) throws Exception {
                //先转为jsonObject
                JSONObject jsonObj = JSON.parseObject(jsonStr);
                JSONObject commonObj = jsonObj.getJSONObject("common");
                Long ts = jsonObj.getLong("ts");
                String hour = DateFormatUtil.toYmdHms(ts).substring(11, 13);
                TrafficHourPageViewBean pageViewBean = new TrafficHourPageViewBean(
                        "",
                        "",
                        hour,
                        0L,
                        0L,
                        1L,
                        ts
                );
                String lastPageId = commonObj.getString("last_page_id");
                if (StringUtils.isEmpty(lastPageId)) {
                    pageViewBean.setSvCt(1L);
                }
                out.collect(pageViewBean);
            }
        });
        //独立访客
        SingleOutputStreamOperator<TrafficHourPageViewBean> uvBeanDS = uvStrDS.process(new ProcessFunction<String, TrafficHourPageViewBean>() {
            @Override
            public void processElement(String jsonStr, ProcessFunction<String, TrafficHourPageViewBean>.Context ctx, Collector<TrafficHourPageViewBean> out) throws Exception {
                //转为json
                JSONObject jsonObj = JSON.parseObject(jsonStr);
                Long ts = jsonObj.getLong("ts");
                String hour = DateFormatUtil.toYmdHms(ts).substring(11, 13);
                TrafficHourPageViewBean pageViewBean = new TrafficHourPageViewBean(
                        "",
                        "",
                        hour,
                        1L,
                        0L,
                        0L,
                        ts
                );
                out.collect(pageViewBean);
            }
        });
        //TODO 5.将两条流union
        DataStream<TrafficHourPageViewBean> unionDS = uvBeanDS.union(pageViewBeanDS);
        //TODO 6.设置watermark 指定事件时间字段
        SingleOutputStreamOperator<TrafficHourPageViewBean> withWatermarkDS = unionDS.assignTimestampsAndWatermarks(WatermarkStrategy.<TrafficHourPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<TrafficHourPageViewBean>() {
                    @Override
                    public long extractTimestamp(TrafficHourPageViewBean pageViewBean, long recordTimestamp) {
                        return pageViewBean.getTs();
                    }
                }));
        //TODO 7.按照小时进行分组
        KeyedStream<TrafficHourPageViewBean, String> keyedDS = withWatermarkDS.keyBy(data -> data.getHour());
        //TODO 8.开窗
        WindowedStream<TrafficHourPageViewBean, String, TimeWindow> windowDS = keyedDS.window(TumblingEventTimeWindows.of(Time.seconds(10)));
        //TODO 9.聚合
        SingleOutputStreamOperator<TrafficHourPageViewBean> reduceDS = windowDS.reduce(new ReduceFunction<TrafficHourPageViewBean>() {
                                                                                           @Override
                                                                                           public TrafficHourPageViewBean reduce(TrafficHourPageViewBean value1, TrafficHourPageViewBean value2) throws Exception {
                                                                                               value1.setUvCt(value1.getUvCt() + value1.getUvCt());
                                                                                               value1.setPvCt(value1.getPvCt() + value1.getPvCt());
                                                                                               value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                                                                                               return value1;
                                                                                           }
                                                                                       },
                new WindowFunction<TrafficHourPageViewBean, TrafficHourPageViewBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TrafficHourPageViewBean> input, Collector<TrafficHourPageViewBean> out) throws Exception {
                        for (TrafficHourPageViewBean trafficHourPageViewBean : input) {
                            trafficHourPageViewBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                            trafficHourPageViewBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                            trafficHourPageViewBean.setTs(System.currentTimeMillis());
                            out.collect(trafficHourPageViewBean);
                        }
                    }
                });
        reduceDS.print(">>>>");
        //TODO 10.将聚合的结果写入clickhouse
        reduceDS.addSink(MyClickhouseUtil.getSinkFunction("insert into dws_traffic_hour_page_view_window  values(?,?,?,?,?,?,?)"));
        env.execute();
    }
}
